package com.learn.datasource;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @create: 2023-04-17 23:22
 * @author: Mr.Du
 * --------------
 * @notes: 基于Socket的Source
 **/
public class StreamSocketSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        int parallelism = env.getParallelism();

        System.out.println("执行环境的默认并行度为：" + parallelism);

        DataStreamSource<String> lines = env.socketTextStream("node1", 9999);

        System.out.println("SocketSource的并行度为：" + lines.getParallelism());

        SingleOutputStreamOperator<String> flatMapWordCount = lines.flatMap((String in, Collector<String> out) -> {
            String[] words = in.split(" ");
            for (String word : words) {
                out.collect(word);
            }
        });
        System.out.println("flatMap的并行度为：" + flatMapWordCount.getParallelism());
        flatMapWordCount.returns(Types.STRING)
                .map(word-> Tuple2.of(word,1)).returns(Types.TUPLE(Types.STRING,Types.INT))
                .keyBy(0).sum(1).print();

        env.execute("");
    }
}
